\newpage

\section{Related Work}
Performing joins on data streams has been the focus of much research on streams \cite{...}. Existing works fall into two branches: one focuses on algorithm design and optimization for stream join processing on single machine, and the other focuses on large-scale distributed stream join processing.

\subsection{Stream Join on Single Machine}

\subsubsection{Centralized Algorithms}
Most of the classical stream join algorithms apply the centralized maintenance of the join states. They can be classified in two groups: hashing based and sorting based.
%nested-loop-based.

\paragraph*{Hashing-based Join}
One of the pioneer algorithms of hashing-based stream join is the pipelining hash-join algorithm~\cite{pdis91:Wilschut}. Nevertheless, it is not practical since it requires that the main memory must be large enough to keep the entire join state. To address this problem, the double-pipelined-hash-join~\cite{sigmod99:Ives}, XJoin~\cite{vldb01:Urhan} and hash-merge join~\cite{icde04:Mokbel} allow parts of the hash table to be flushed to disk for later processing. In order to maximize the output rate, the state-of-the-art RPJ~\cite{sigmod05:Tao} further applies a statistics-based flushing policy that exploits the statistics of tuple join probability to keep in memory those tuples that are more likely to join. 

\paragraph*{Sorting-based Join}
The inherence of hashing-based join processing make that family of algorithms insufficient to handle inequality join. Alternatively, sorting-based join algorithms can support inequality joins, but they have been traditionally deemed inappropriate for stream joins because sorting is a blocking operation that requires seeing the entire input before producing any output, which contradicts the unbounded inputs. To circumvent this problem, the progressive merge join~\cite{vldb02:Dittrich} is developed to be sorting-based but non-blocking. However, it produces no join results until the memory gets filled, resulting in high delay of the outputs.

%\paragraph*{Nested-loop-based Join}
%Nested-loop-based join algorithms~\cite{book:Ullman} are often deemed naive in relational database, but they are revisited in stream join processing. \cite{tkde10:Bornea} develops double index nested-loops reactive join an intuitive flushing policy that aims to increase the productivity of in-memory tuples in producing results during the online phase of the join, and a novel reentrant join technique that allows the algorithm to rapidly switch between processing in-memory and disk-resident tuples, thus, better exploiting temporary delays when new data are not available.

\subsubsection{Multi-core Compatible Algorithms}
Modern multi-core technology brings parallelism into single machine processing capability. \cite{vldb07:Gedik} designs the windowed stream join operators on multi-core Cell processor. But its efficiency highly relies on the hardware parallelism provided by the Cell processor, making it little general to commodity hardware.
%Instead of maintaining the centralized states of join processing, 
\cite{sigmod11:Teubner} proposes the handshake join idea for processing stream joins based on multi-core and shared memory. It is characterized for decentralized dataflow-oriented processing and autonomic load balancing. It adopted the tuple-nested-loop join processing that carries out the brute-force pairwise comparison on each tuple against all tuples in the opposite stream, leading to poor processing efficiency. Further, it is highly customized for single machine with modern multi-core hardware and share-memory paradigm, which could reversely be its limitation of scalability, especially when being applied in the distributed enviroment. ESJ enriches this model based on message passing, and enhances the processing for theta-join. By piggybacking the resolution of relational operator with the dataflow processing, ESJ supports both equality and inequality joins without the sorting-based subroutine. Meanwhile, ESJ's real-time join window control guarantees the timeliness of join outputs.

\subsection{Distributed Stream Join}
Single-machine stream join often cannot scale to effectively handle big streaming inputs due to the capacity constraint of shared-memory. Consequently, scalable processing of stream join in the distributed environment becomes attractive. 
%Some recent researches on distributed stream join processing~\cite{...} provide solutions with respect to effectiveness and efficiency.
Photon~\cite{sigmod13:Ananthanarayanan} is a fault-tolerant, scalable system optimized for joining data streams of web search queries and user clicks on advertisements. However, it is an ad hoc system, exhibiting little generality for porting to various applications.
D-Stream~\cite{hotcloud12:Zaharia} breaks continuous streams into discrete units and processes them as batch jobs on Spark~\cite{nsdi12:Zaharia}. Such treating a streaming computation as a series of batch computations is not acceptable for stream join in ubiquitous applications since some tuple pairs in separated mini-batch jobs may need to join but would be missed-join. 
Similarly, \cite{sigmod10:Blanas} uses MapReduce~\cite{osdi04:Dean} to join streams, which is more suitable for a batch system rather than a continuously running one.
\cite{eurosys13:Qian} exploits the tuple dependency to conduct the stream join on MapReduce-like system. But maintaining the dependencies is centralized, making it tend to be the performance bottleneck. Moreover, the dependency resolving would become very complicated when dealing with multiple join predicates; its optimization relies on the application-awareness.
\cite{edbt09:Wang} proposes an operator distribution scheme that separates a macro join operator into a series of smaller operators by time-slicing of the states. The processing is distributed to these sub-operators which are connected in a ring architecture. The performance issue results from the state synchronization, which logically requires centralized maintenance. Although it makes a special design of synchronization based on multicasting, it incurs high communication overhead which could be exponential to the number of sub-operators, leading to limitation of scalability.

Our proposed ESJ is a general operator for distributed stream join. ESJ applies the stateless stream join processing and the decentralized join window control based on the dataflow-oriented model, making it highly scalable in the distributed environment.
